package rxgo

import (
	"context"
	"reflect"
	"sync"
	"time"
)

type DebounceKindOperator struct {
	duration time.Duration
	opFunc   func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

type SampleKindOperator struct {
	duration time.Duration
	opFunc   func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

func (tsop DebounceKindOperator) op(ctx context.Context, o *Observable) {
	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
	// this resurces may be changed when operation routine is running.
	in := o.pred.outflow
	out := o.outflow
	//fmt.Println(o.name, "operator in/out chan ", in, out)
	var wg sync.WaitGroup

	go func() {
		end := false

		content := []interface{}{}

		start := time.Now()

		x := <-in

		content = append(content, x)

		for x = range in {
			now := time.Now()
			duration := now.Sub(start)
			if (duration) < tsop.duration {
				continue
			} else {
				content = append(content, x)
				start = now
			}

		}

		for _, x := range content {

			if end {
				continue
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			xv := reflect.ValueOf(x)
			// send an error to stream if the flip not accept error

			// scheduler
			switch threading := o.threading; threading {
			case ThreadingDefault:
				if tsop.opFunc(ctx, o, xv, out) {
					end = true
				}
			case ThreadingIO:
				fallthrough
			case ThreadingComputing:
				wg.Add(1)
				go func() {
					defer wg.Done()
					if tsop.opFunc(ctx, o, xv, out) {
						end = true
					}
				}()
			default:
			}
		}

		wg.Wait() //waiting all go-routines completed
		o.closeFlow(out)
	}()
}

func (tsop SampleKindOperator) op(ctx context.Context, o *Observable) {
	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
	// this resurces may be changed when operation routine is running.
	in := o.pred.outflow
	out := o.outflow
	//fmt.Println(o.name, "operator in/out chan ", in, out)
	var wg sync.WaitGroup

	go func() {
		end := false
		arr := []interface{}{}
		timeArr := []time.Time{}
		for x := range in {
			arr = append(arr, x)
			timeArr = append(timeArr, time.Now())
		}

		count := 0
		var x interface{}
		for index, timee := range timeArr {

			if len(timeArr) == 0 {
				break
			}

			durationTime := timee.Sub(timeArr[0])

			periodNum := (durationTime - durationTime%tsop.duration) / tsop.duration

			println("periodNum:", periodNum)

			if int(periodNum) == count && index != len(timeArr)-1 {
				x = arr[index]
				continue
			} else {
				count = int(periodNum)
				tsop.opFunc(ctx, o, reflect.ValueOf(x), out)
				x = arr[index]
				if index == len(timeArr)-1 {
					tsop.opFunc(ctx, o, reflect.ValueOf(x), out)
				}
			}

			if end {
				continue
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!

		}

		wg.Wait() //waiting all go-routines completed
		o.closeFlow(out)
	}()
}

type LastKindOperator struct {
	num    int
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
	mode   int // last or last n
}

func (tsop LastKindOperator) op(ctx context.Context, o *Observable) {
	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
	// this resurces may be changed when operation routine is running.
	in := o.pred.outflow
	out := o.outflow
	//fmt.Println(o.name, "operator in/out chan ", in, out)
	var wg sync.WaitGroup

	go func() {
		end := false
		arr := []interface{}{}
		for x := range in {
			arr = append(arr, x)
		}

		for index, x := range arr {
			if tsop.mode == 0 && index != len(arr)-1 {
				continue
			}
			if tsop.mode == 1 && index < len(arr)-tsop.num {
				continue
			}
			if tsop.mode == 2 && index >= len(arr)-tsop.num {
				continue
			}
			if tsop.mode == 3 {
				_, ok := x.(error)
				if ok {
					_ = tsop.opFunc(ctx, o, reflect.ValueOf(x.(error)), out)
					continue
				} else {
					continue
				}
			}
			if end {
				continue
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			xv := reflect.ValueOf(x)
			// send an error to stream if the flip not accept error

			// scheduler
			switch threading := o.threading; threading {
			case ThreadingDefault:
				if tsop.opFunc(ctx, o, xv, out) {
					end = true
				}
			case ThreadingIO:
				fallthrough
			case ThreadingComputing:
				wg.Add(1)
				go func() {
					defer wg.Done()
					if tsop.opFunc(ctx, o, xv, out) {
						end = true
					}
				}()
			default:
			}
		}

		wg.Wait() //waiting all go-routines completed
		o.closeFlow(out)
	}()
}

func (parent *Observable) Debounce(duration time.Duration) (o *Observable) {
	// check validation of f

	o = parent.newTransformObservable("Debounce")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = nil
	o.operator = DebounceKindOperator{duration, func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

		_ = o.sendToFlow(ctx, x.Interface(), out)
		return false
	}}
	return o
}

func (parent *Observable) Sample(duration time.Duration) (o *Observable) {
	// check validation of f

	o = parent.newTransformObservable("Sample")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = nil

	o.operator = SampleKindOperator{duration, func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

		_ = o.sendToFlow(ctx, x.Interface(), out)
		return false
	}}
	return o
}

var FirstOperator = transOperater{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	end = false

	// send data
	if !end {
		_ = o.sendToFlow(ctx, x.Interface(), out)
		end = true
	}

	return
}}

func (parent *Observable) First() (o *Observable) {
	// check validation of f

	o = parent.newTransformObservable("first")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = nil
	o.operator = FirstOperator
	return o
}

var DistinctOperator = transOperater{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	temp := false
	if o.flip.(func(interface{}) bool)(x.Interface()) {
		temp = true
	}

	end = false

	// send data
	if !end && temp {
		_ = o.sendToFlow(ctx, x.Interface(), out)

	}
	return
}}

//Distinct param cant be slice type
func (parent *Observable) Distinct() (o *Observable) {

	o = parent.newTransformObservable("distinct")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = func() func() func(interface{}) bool {
		return func() func(interface{}) bool {
			set := map[interface{}]bool{}
			return func(item interface{}) bool {

				_, exists := set[item]
				if exists || set[item] {
					return false
				} else {
					set[item] = true
					return true
				}
			}
		}
	}()()
	o.operator = DistinctOperator
	return o
}

var ElementAtOperator = transOperater{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	temp := o.flip.(func() bool)()

	end = false

	// send data
	if !end && temp {
		_ = o.sendToFlow(ctx, x.Interface(), out)

	}
	return
}}

func (parent *Observable) ElementAt(index int) (o *Observable) {

	o = parent.newTransformObservable("elementat")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = func() func() func() bool {
		return func() func() bool {
			indexx := 0
			max_index := index
			return func() bool {
				if indexx < max_index {
					indexx++
					return false
				}
				if indexx == max_index {
					indexx++
					return true
				}
				if indexx > max_index {
					return false
				}
				return false
			}
		}
	}()()
	o.operator = ElementAtOperator
	return o
}

var SkipOperator = transOperater{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	temp := o.flip.(func() bool)()

	end = false

	// send data
	if !end && temp {
		_ = o.sendToFlow(ctx, x.Interface(), out)

	}
	return
}}

func (parent *Observable) Skip(index int) (o *Observable) {

	o = parent.newTransformObservable("Skip")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = func() func() func() bool {
		return func() func() bool {
			indexx := 0
			max_index := index
			return func() bool {
				if indexx < max_index {
					indexx++
					return false
				}
				if indexx == max_index {
					indexx++
					return true
				}
				if indexx > max_index {
					return true
				}
				return false
			}
		}
	}()()
	o.operator = SkipOperator
	return o
}

var TakeOperator = transOperater{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	temp := o.flip.(func() bool)()

	end = false

	// send data
	if !end && temp {
		_ = o.sendToFlow(ctx, x.Interface(), out)

	}
	return
}}

func (parent *Observable) Take(index int) (o *Observable) {

	o = parent.newTransformObservable("Take")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = func() func() func() bool {
		return func() func() bool {
			indexx := 0
			max_index := index
			return func() bool {
				if indexx < max_index {
					indexx++
					return true
				}
				if indexx == max_index {
					indexx++
					return false
				}
				if indexx > max_index {
					return false
				}
				return false
			}
		}
	}()()
	o.operator = TakeOperator
	return o
}

var LastOperater = LastKindOperator{0, func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	_ = o.sendToFlow(ctx, x.Interface(), out)
	return false
}, 0}

func (parent *Observable) Last() (o *Observable) {

	// check validation of f

	o = parent.newTransformObservable("Last")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = nil
	o.operator = LastOperater
	return o

}

func (parent *Observable) TakeLast(n int) (o *Observable) {
	o = parent.newTransformObservable("TakeLast")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = nil
	o.operator = LastKindOperator{n, func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

		_ = o.sendToFlow(ctx, x.Interface(), out)
		return false
	}, 1}
	return o

}

func (parent *Observable) SkipLast(n int) (o *Observable) {
	o = parent.newTransformObservable("TakeLast")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = nil
	o.operator = LastKindOperator{n, func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

		_ = o.sendToFlow(ctx, x.Interface(), out)
		return false
	}, 2}
	return o

}

var IgnoreElementOperater = LastKindOperator{0, func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	_ = o.sendToFlow(ctx, x.Interface(), out)

	return false
}, 3}

func (parent *Observable) IgnoreElement() (o *Observable) {

	o = parent.newTransformObservable("IgnoreElement")
	o.flip_accept_error = false

	o.flip_sup_ctx = false
	o.flip = nil
	o.operator = IgnoreElementOperater

	return o

}

/*
var filterOperater = transOperater{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	fv := reflect.ValueOf(o.flip)
	var params = []reflect.Value{x}
	rs, skip, stop, e := userFuncCall(fv, params)

	var item interface{} = rs[0].Interface()
	if stop {
		end = true
		return
	}
	if skip {
		return
	}
	if e != nil {
		item = e
	}
	// send data
	if !end {
		end = o.sendToFlow(ctx, item, out)
	}

	return
}}
var filterOperater = transOperater{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {

	fv := reflect.ValueOf(o.flip)
	var params = []reflect.Value{x}
	rs, skip, stop, e := userFuncCall(fv, params)

	var item interface{} = rs[0].Interface()
	if stop {
		end = true
		return
	}
	if skip {
		return
	}
	if e != nil {
		item = e
	}
	// send data
	if !end {
		if b, ok := item.(bool); ok && b {
			end = o.sendToFlow(ctx, x.Interface(), out)
		}
	}

	return
}}

func (parent *Observable) filter(f interface{}) (o *Observable) {

	// check validation of f
	fv := reflect.ValueOf(f)
	inType := []reflect.Type{typeAny}
	outType := []reflect.Type{typeAny}
	b, ctx_sup := checkFuncUpcast(fv, inType, outType, true)
	if !b {
		panic(ErrFuncFlip)
	}

	o = parent.newTransformObservable("filter")
	o.flip_accept_error = checkFuncAcceptError(fv)

	o.flip_sup_ctx = ctx_sup
	o.flip = fv.Interface()
	o.operator = filterOperater
	return o

}
func (parent *Observable) Filter(f interface{}) (o *Observable) {
	// check validation of f
	fv := reflect.ValueOf(f)
	inType := []reflect.Type{typeAny}
	outType := []reflect.Type{typeBool}
	b, ctx_sup := checkFuncUpcast(fv, inType, outType, true)
	if !b {
		panic(ErrFuncFlip)
	}

	o = parent.newTransformObservable("filter")
	o.flip_accept_error = checkFuncAcceptError(fv)

	o.flip_sup_ctx = ctx_sup
	o.flip = fv.Interface()
	o.operator = filterOperater
	return o
}
*/
